package com.homedo.rxj;

import com.homedo.distributed.id.client.gen.constants.GenExecutorConstants;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RxjTest7 {
    static AtomicInteger count = new AtomicInteger();
    static ObservableEmitter<Integer> curEmitter = null;

    private static ExecutorService updateBufferService = GenExecutorConstants.MONITOR_BUFFER_EXECUTOR;


    public static void main(String[] args) throws InterruptedException {
        Observable<Integer> observable = createObserver();

//        observable.subscribeOn(Schedulers.from(updateBufferService)).observeOn(Schedulers.from(updateBufferService));

        observable.buffer(3).subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> o) throws Exception {
                System.out.println(Thread.currentThread() + "   " + o);
            }
        });

//        Disposable disposable = observable.subscribe(new Consumer<Integer>() {
//            @Override
//            public void accept(Integer o) throws Exception {
//                System.out.println(Thread.currentThread() + "   " + o);
//            }
//        });

        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10000; i++) {
                    count.incrementAndGet();

                    if (new Random().nextInt(20) % 10 == 0) {
                        try {
                            TimeUnit.MILLISECONDS.sleep(500);
//                        System.out.println(Thread.currentThread()+"000");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                }
            }
        });

//        thread.setDaemon(true);  //不设置成后台线程 interval会一直执行
        thread.start();

        System.out.println("====test===");

//        TimeUnit.SECONDS.sleep(10);
//        curEmitter.onComplete();
        System.out.println("====onComplete===");

    }


    private static Observable<Integer> createObserver() {
        Observable.interval(1, 1, TimeUnit.SECONDS, Schedulers.from(updateBufferService)).subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                if (!curEmitter.isDisposed()) {
                    curEmitter.onNext(count.get());// 延迟1秒后 每隔1秒发送一下统计数据
                    count.set(0);//重置统计数据
//                System.out.println(Thread.currentThread()+"--");
                }
                System.out.println(aLong);
            }
        });


        return Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        curEmitter = emitter;
                    }
                }
        ).subscribeOn(Schedulers.from(updateBufferService));

    }
}
